package psnl.bingo.mr.demo4;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

public class D4RecordWriter extends RecordWriter<D4Line, NullWritable> {
    public static String outPathDir = null;
    FSDataOutputStream save2020 = null;
    FSDataOutputStream save2021 = null;
    FSDataOutputStream saveOther = null;

    public D4RecordWriter(TaskAttemptContext job) {
        try {
            FileSystem fs = FileSystem.get(job.getConfiguration());

            save2020 = fs.create(new Path(outPathDir+"/2020.txt"));
            save2021 = fs.create(new Path(outPathDir+"/2021.txt"));
            saveOther = fs.create(new Path(outPathDir+"/other.txt"));

        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void write(D4Line key, NullWritable value) throws IOException, InterruptedException {
        if(key.getMonth().startsWith("2020")) {
            save2020.writeBytes(key+"\n");
        }else if(key.getMonth().startsWith("2021")) {
            save2021.writeBytes(key+"\n");
        }else {
            saveOther.writeBytes(key+"\n");
        }
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        IOUtils.closeStream(save2020);
        IOUtils.closeStream(save2021);
        IOUtils.closeStream(saveOther);
    }
}
